package com.we.risk.phoneRecord.stage1;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class PhoneRecordLabelAddBakKfkSrc {
    public static final String RELEASEPROP = "risk/phonerecord/phonerecord_prod.properties";
    //    public static final String RELEASEPROP = "risk/phonerecord/test.properties";
    public static final String PROPSETS = "risk/phonerecord/phonerecord220822_s1.properties";
    public static Logger LOG = LoggerFactory.getLogger(PhoneRecordLabelAddBakKfkSrc.class);

    public static final String KFKBROKERHOSTBAK =
            "10.10.13.22:9092,10.10.13.23:9092,10.10.13.24:9092";
    public static final String KFKBAKTOPIC = "kafka.topic.data.record";
    //    public static final String KFKBROKERHOSTBAK =
    // "172.16.1.41:9092,172.16.1.42:9092,172.16.1.43:9092,172.16.1.44:9092";
    //    public static final String KFKBAKTOPIC = "risk_phone_record_data_source2";
    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String MSGTYPE = "HFQ_UPLOAD_PHONE_CALL_RECORD";
    public static final String DATEEARLIEST = "2000-01-01 00:00:00";
    public static final int MAXDAYS = 15000;
    public static final int[] TIMESECTION = {31, 91, 181, 271, 361, 721, MAXDAYS};
    public static SimpleDateFormat sdf = new SimpleDateFormat(DATEFORMAT);
    public static final String KEYBY = "user_key";
    public static final int LABELSIZE = 739;

    public static final String ALLPRLISTS = "continous"; // _call  or _uncall or _uncall3days
    public static final String MOBILEPRLISTS = "mobile_continous";
    public static final String MOBILEINBOOKOUTGOPRLISTS = "mobile_inbook_outgo_continous";
    public static final String MOBILEINBOOKINCOMELISTS = "mobile_inbook_income_continous";
    public static final String MOBILEINBOOKREJECTEDPRLISTS = "mobile_inbook_rejected_continous";
    public static final String MOBILEINBOOKMISSEDPRLISTS = "mobile_inbook_missed_continous";
    public static final String MOBILEINBOOKBLOCKEDPRLISTS = "mobile_inbook_blocked_continous";
    public static final String MOBILEINBOOKUNKNOWNPRLISTS = "mobile_inbook_unknown_continous";
    public static final String MOBILENOTINBOOKOUTGOPRLISTS = "mobile_notbook_outgo_continous";
    public static final String MOBILENOTINBOOKINCOMELISTS = "mobile_notbook_income_continous";
    public static final String MOBILENOTINBOOKREJECTEDPRLISTS = "mobile_notbook_rejected_continous";
    public static final String MOBILENOTINBOOKMISSEDPRLISTS = "mobile_notbook_missed_continous";
    public static final String MOBILENOTINBOOKBLOCKEDPRLISTS = "mobile_notbook_blocked_continous";
    public static final String MOBILENOTINBOOKUNKNOWNPRLISTS = "mobile_notbook_unknown_continous";
    public static final String FIXPHONEPRLISTS = "fix_continous";
    public static final String FIXPHONEINBOOKOUTGOPRLISTS = "fix_inbook_outgo_continous";
    public static final String FIXPHONEINBOOKINCOMEPRLISTS = "fix_inbook_income_continous";
    public static final String FIXPHONEINBOOKREJECTEDPRLISTS = "fix_inbook_rejected_continous";
    public static final String FIXPHONEINBOOKMISSEDPRLISTS = "fix_inbook_missed_continous";
    public static final String FIXPHONEINBOOKBLOCKEDPRLISTS = "fix_inbook_blocked_continous";
    public static final String FIXPHONEINBOOKUNKNOWNPRLISTS = "fix_inbook_unknown_continous";
    public static final String FIXPHONENOTINBOOKOUTGOPRLISTS = "fix_notbook_outgo_continous";
    public static final String FIXPHONENOTINBOOKINCOMEPRLISTS = "fix_notbook_income_continous";
    public static final String FIXPHONENOTINBOOKREJECTEDPRLISTS = "fix_notbook_rejected_continous";
    public static final String FIXPHONENOTINBOOKMISSEDPRLISTS = "fix_notbook_missed_continous";
    public static final String FIXPHONENOTINBOOKBLOCKEDPRLISTS = "fix_notbook_blocked_continous";
    public static final String FIXPHONENOTINBOOKUNKNOWNPRLISTS = "fix_notbook_unknown_continous";
    public static final String PHONEBOOKPRLISTS = "inbook_continous";
    public static final String NOTPHONEBOOKPRLISTS = "notbook_continous";
    public static final String OUTGOINGPRLISTS = "outgo_continous";
    public static final String INCOMINGPRLISTS = "income_continous";
    public static final String REJECTEDPRLISTS = "rejected_continous";
    public static final String MISSEDPRLISTS = "missed_continous";
    public static final String BLOCKEDPRLISTS = "blocked_continous";
    public static final String UNKNOWNPRLISTS = "unknown_continous";

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.latest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            consumProp.setProperty("group.id", srckfkGrupId + "bak");
            KafkaSource<String> kfkBakSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(KFKBROKERHOSTBAK)
                            .setTopics(KFKBAKTOPIC)
                            .setStartingOffsets(OffsetsInitializer.latest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "PhoneRecord")
                            .uid("kfk-PhoneRecordLabel-source");
            SingleOutputStreamOperator<String> inputBak =
                    env.fromSource(
                                    kfkBakSource,
                                    WatermarkStrategy.noWatermarks(),
                                    "PhoneRecord-bak")
                            .uid("kfk-PhoneRecordLabel-bak-source");

            DataStream<String> unionInput = input.union(inputBak);

            SingleOutputStreamOperator<String> out =
                    unionInput
                            .flatMap(
                                    new RichFlatMapFunction<String, String>() {
                                        @Override
                                        public void flatMap(String input, Collector<String> out)
                                                throws Exception {
                                            long now = System.currentTimeMillis();
                                            try {
                                                if (input.contains(MSGTYPE)) {
                                                    JSONObject inputJson =
                                                            JSONObject.parseObject(input);
                                                    if (inputJson != null
                                                            && inputJson
                                                                    .getString("type")
                                                                    .equals(MSGTYPE)) {
                                                        JSONObject dataJson =
                                                                inputJson.getJSONObject("dataJson");
                                                        Long captureTime =
                                                                dataJson.getLong("timestamp");
                                                        JSONObject job =
                                                                dataJson.getJSONObject("job");
                                                        String jobID = job.getString("jobID");

                                                        String userKey =
                                                                dataJson.getString("userKey");
                                                        JSONObject data =
                                                                dataJson.getJSONObject("data");
                                                        JSONArray calls =
                                                                data.getJSONArray("phoneCallList");
                                                        int size = calls.size();
                                                        /**
                                                         * String captureDate = sdf.format(new
                                                         * Date(captureTime));
                                                         * System.out.println("captureTime: " +
                                                         * captureTime);
                                                         * System.out.println("captureDate: " +
                                                         * captureDate);
                                                         * System.out.println("userKey: " +
                                                         * userKey); System.out.println("call list
                                                         * size: " + size);
                                                         */
                                                        Map<String, Set<String>> prlists =
                                                                initPRlists();
                                                        /** add list */
                                                        for (int i = 0; i < size; i++) {
                                                            JSONObject call =
                                                                    (JSONObject) calls.get(i);
                                                            String callType =
                                                                    call.getString("type");
                                                            String callName =
                                                                    call.getString("name");
                                                            boolean isInPhoneBook = false;
                                                            if (callName != null
                                                                    && !callName.equals("")) {
                                                                isInPhoneBook = true;
                                                            }
                                                            String callNumber =
                                                                    call.getString("number");
                                                            boolean isMobile = isMobile(callNumber);
                                                            Long callTime =
                                                                    Long.valueOf(
                                                                            call.getString("date"));
                                                            String callDate =
                                                                    sdf.format(new Date(callTime));

                                                            //
                                                            //      System.out.println("" + callType
                                                            // + (isInPhoneBook ? " inbook" : "
                                                            // notbook")
                                                            //
                                                            //              + (isMobile ? " mobile "
                                                            // : " fix ")
                                                            //
                                                            //              + "callDate: " +
                                                            // callDate);

                                                            if (captureTime >= callTime
                                                                    && callTime
                                                                            >= sdf.parse(
                                                                                            DATEEARLIEST)
                                                                                    .getTime()) {
                                                                prlists.get(ALLPRLISTS)
                                                                        .add(callDate);
                                                                addPrlist(
                                                                        prlists,
                                                                        callType,
                                                                        PrType.DEFAULT,
                                                                        callDate);
                                                                if (isMobile) {
                                                                    prlists.get(MOBILEPRLISTS)
                                                                            .add(callDate);
                                                                    if (isInPhoneBook) {
                                                                        addPrlist(
                                                                                prlists,
                                                                                callType,
                                                                                PrType.MOBILEINBOOK,
                                                                                callDate);
                                                                    } else {
                                                                        addPrlist(
                                                                                prlists,
                                                                                callType,
                                                                                PrType
                                                                                        .MOBILENOTINBOOK,
                                                                                callDate);
                                                                    }

                                                                } else {
                                                                    prlists.get(FIXPHONEPRLISTS)
                                                                            .add(callDate);
                                                                    if (isInPhoneBook) {
                                                                        addPrlist(
                                                                                prlists,
                                                                                callType,
                                                                                PrType.FIXINBOOK,
                                                                                callDate);
                                                                    } else {
                                                                        addPrlist(
                                                                                prlists,
                                                                                callType,
                                                                                PrType.FIXNOTINBOOK,
                                                                                callDate);
                                                                    }
                                                                }
                                                                if (isInPhoneBook) {
                                                                    prlists.get(PHONEBOOKPRLISTS)
                                                                            .add(callDate);
                                                                } else {
                                                                    prlists.get(NOTPHONEBOOKPRLISTS)
                                                                            .add(callDate);
                                                                }
                                                            }
                                                        }

                                                        JSONObject outJson = new JSONObject();
                                                        outJson.put(KEYBY, userKey);
                                                        outJson.put("capture_time", captureTime);
                                                        outJson.put("job_id", jobID);

                                                        /** calc label */
                                                        Set<Map.Entry<String, Set<String>>>
                                                                entries = prlists.entrySet();
                                                        for (Map.Entry<String, Set<String>>
                                                                element : entries) {
                                                            String key = element.getKey();
                                                            Set<String> sets = element.getValue();
                                                            //
                                                            //      int listSize = sets.size();
                                                            //
                                                            //      System.out.println("##" + key +
                                                            // " size: " + listSize);
                                                            ArrayList<String> lists =
                                                                    new ArrayList<>();
                                                            lists.addAll(sets);
                                                            Collections.sort(lists);
                                                            calcFeature(
                                                                    lists,
                                                                    captureTime,
                                                                    key,
                                                                    outJson);

                                                            /**
                                                             * Collections.reverse(lists);
                                                             * Iterator<String> iterator =
                                                             * lists.iterator();
                                                             * while(iterator.hasNext()) {
                                                             * System.out.print(iterator.next() +
                                                             * ","); } System.out.println("");
                                                             */
                                                        }

                                                        long end = System.currentTimeMillis();
                                                        outJson.put("label_output_time", end);
                                                        LOG.debug(
                                                                " costTime: "
                                                                        + (end - now)
                                                                        + " ms"
                                                                        + " end-cap time: "
                                                                        + (end - captureTime)
                                                                        + " ms");

                                                        if (outJson.containsKey(KEYBY)
                                                                && (outJson.getString(KEYBY)
                                                                        != null)
                                                                && outJson.size() == LABELSIZE) {
                                                            out.collect(outJson.toString());
                                                        } else {
                                                            LOG.error(
                                                                    "OutJson erorr:"
                                                                            + outJson.toString());
                                                            System.out.println(
                                                                    "OutJson eror:"
                                                                            + outJson.toString());
                                                        }
                                                    }
                                                }
                                            } catch (Exception e) {
                                                LOG.error(e.getMessage());
                                                LOG.error("Flatmap1 Exception Input: " + input);
                                                System.out.println(e.getMessage());
                                                System.out.println(
                                                        "Flatmap1 Exception Input: " + input);
                                                e.printStackTrace();
                                            }
                                        }
                                    })
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject jsonObject = JSONObject.parseObject(value);
                                            return jsonObject.getString(KEYBY);
                                        }
                                    })
                            .flatMap(
                                    new RichFlatMapFunction<String, String>() {
                                        ValueState<JSONObject> phonerecordLabel;

                                        @Override
                                        public void open(Configuration parameters)
                                                throws Exception {
                                            phonerecordLabel =
                                                    getRuntimeContext()
                                                            .getState(
                                                                    new ValueStateDescriptor<
                                                                            JSONObject>(
                                                                            "phonerecordLabel",
                                                                            JSONObject.class));
                                        }

                                        @Override
                                        public void flatMap(String value, Collector<String> out)
                                                throws Exception {
                                            try {
                                                JSONObject inputJson =
                                                        JSONObject.parseObject(value);
                                                JSONObject tmpState = phonerecordLabel.value();
                                                JSONObject curState =
                                                        tmpState == null
                                                                ? initPropJson(PROPSETS)
                                                                : tmpState;

                                                String outKey = inputJson.getString(KEYBY);

                                                Iterator<String> iterator =
                                                        inputJson.keySet().iterator();
                                                while (iterator.hasNext()) {
                                                    String key = iterator.next();
                                                    if (!key.equals(KEYBY)
                                                            && curState.containsKey(key)) {
                                                        Object labelV = inputJson.get(key);
                                                        if (!curState.get(key).equals(labelV)) {
                                                            JSONObject outJson = new JSONObject();
                                                            outJson.put(KEYBY, outKey);
                                                            outJson.put("label_name", key);
                                                            outJson.put("label_value", labelV);
                                                            out.collect(outJson.toString());
                                                            curState.put(key, labelV);
                                                        }
                                                    }
                                                }
                                                phonerecordLabel.update(curState);
                                            } catch (Exception e) {
                                                LOG.error(e.getMessage());
                                                LOG.error("Flatmap2 Exception Input: " + value);
                                                System.out.println(e.getMessage());
                                                System.out.println(
                                                        "Flatmap2 Exception Input: " + value);
                                                e.printStackTrace();
                                            }
                                        }
                                    });

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(PhoneRecordLabelAddBakKfkSrc.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(KEYBY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }

    private static JSONObject calcFeature(
            List<String> dates, long captureTime, String prType, JSONObject out) {
        int size = dates.size();
        StringBuilder stringBuilder = new StringBuilder();
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(captureTime);

        for (int i = size - 1; i >= 0; i--) {
            while (true) {
                if (sdf.format(calendar.getTime()).compareTo(dates.get(i)) == 0) {
                    stringBuilder.append("1");
                    calendar.add(Calendar.DATE, -1);
                    break;
                } else {
                    stringBuilder.append("0");
                    calendar.add(Calendar.DATE, -1);
                }
            }
        }
        String input = stringBuilder.toString();
        calcFeatureEx(input, prType, out);

        return out;
    }

    public static void calcFeatureEx(String input, String prType, JSONObject out) {
        if (input != null) {
            int inputLen = input.length();
            String substr = "";
            LOG.debug(prType + " inputLen: " + inputLen);
            LOG.debug("input:" + input);

            for (int lastDays : TIMESECTION) {
                if (inputLen >= lastDays) {
                    substr = input.substring(0, lastDays);
                } else {
                    substr = input;
                }
                LOG.debug(prType + "_lastDays:" + lastDays + " content:" + substr);
                int max1len = 0;
                int max0len = 0;
                int silentNum = 0;
                /** 判断是否含有通话记录 */
                if (input.contains("1")) {
                    if (substr.contains("1")) {
                        int start = substr.indexOf("1");
                        int end = substr.lastIndexOf("1");
                        substr = substr.substring(start, end + 1);
                        LOG.debug(prType + "_lastDays:" + lastDays + "#cut# content:" + substr);
                        String[] split1 = substr.split("0");
                        String[] split0 = substr.split("1");
                        for (String item1 : split1) {
                            int itemLen = item1.length();
                            max1len = max1len > itemLen ? max1len : itemLen;
                        }
                        for (String item0 : split0) {
                            int itemLen = item0.length();
                            max0len = max0len > itemLen ? max0len : itemLen;
                            if (itemLen >= 2) {
                                silentNum++;
                            }
                        }
                        /** 13-15 非连续通话记录天数为3 */
                        if (start != end) {
                            max0len++;
                        }
                        /** 只有1天通话记录时 非连续通话记录最大天数的标签按0输出 */
                        //                        else {
                        //                            max0len = -999;
                        //                        }
                    } else {
                        max0len = -999;
                    }
                } else {
                    /** 没有符合该类型的通话记录 连续/非连续类的天数标签均为-999缺失 */
                    max0len = -999;
                    max1len = -999;
                }

                if (lastDays == TIMESECTION[0]) {
                    out.put(prType + "_call_days_1m", max1len);
                    out.put(prType + "_uncall_days_1m", max0len);
                    out.put(prType + "_uncall_2days_num_1m", silentNum);
                } else if (lastDays == TIMESECTION[1]) {
                    out.put(prType + "_call_days_3m", max1len);
                    out.put(prType + "_uncall_days_3m", max0len);
                    out.put(prType + "_uncall_2days_num_3m", silentNum);
                } else if (lastDays == TIMESECTION[2]) {
                    out.put(prType + "_call_days_6m", max1len);
                    out.put(prType + "_uncall_days_6m", max0len);
                    out.put(prType + "_uncall_2days_num_6m", silentNum);
                } else if (lastDays == TIMESECTION[3]) {
                    out.put(prType + "_call_days_9m", max1len);
                    out.put(prType + "_uncall_days_9m", max0len);
                    out.put(prType + "_uncall_2days_num_9m", silentNum);
                } else if (lastDays == TIMESECTION[4]) {
                    out.put(prType + "_call_days_12m", max1len);
                    out.put(prType + "_uncall_days_12m", max0len);
                    out.put(prType + "_uncall_2days_num_12m", silentNum);
                } else if (lastDays == TIMESECTION[5]) {
                    out.put(prType + "_call_days_24m", max1len);
                    out.put(prType + "_uncall_days_24m", max0len);
                    out.put(prType + "_uncall_2days_num_24m", silentNum);
                } else {
                    out.put(prType + "_call_days_his", max1len);
                    out.put(prType + "_uncall_days_his", max0len);
                    out.put(prType + "_uncall_2days_num_his", silentNum);
                }
            }
        }
    }

    private static boolean isMobile(String phone) {
        //        String regex =
        // "^((13[0-9])|(14[5,7,9])|(15([0-3]|[5-9]))|(166)|(17[0,1,3,5,6,7,8])|(18[0-9])|(19[8|9]))\\d{8}$";
        String regex = "1[3-9][0-9]{9}$";
        if (phone != null) {
            phone = phone.replaceAll("[^\\d]", "").replaceAll("^0*(86)*0*", "");
            Pattern p = Pattern.compile(regex);
            Matcher m = p.matcher(phone);
            return m.find();
        } else {
            return false;
        }
    }

    private static void addPrlist(
            Map<String, Set<String>> prlists, String callType, PrType prType, String date) {
        switch (callType) {
                /** 外呼/主叫 */
            case "OUTGOING":
                if (prType == PrType.MOBILEINBOOK) {
                    prlists.get(MOBILEINBOOKOUTGOPRLISTS).add(date);
                } else if (prType == PrType.MOBILENOTINBOOK) {
                    prlists.get(MOBILENOTINBOOKOUTGOPRLISTS).add(date);
                } else if (prType == PrType.FIXINBOOK) {
                    prlists.get(FIXPHONEINBOOKOUTGOPRLISTS).add(date);
                } else if (prType == PrType.FIXNOTINBOOK) {
                    prlists.get(FIXPHONENOTINBOOKOUTGOPRLISTS).add(date);
                } else {
                    prlists.get(OUTGOINGPRLISTS).add(date);
                }
                break;
                /** 内呼/被叫 */
            case "INCOMING":
                if (prType == PrType.MOBILEINBOOK) {
                    prlists.get(MOBILEINBOOKINCOMELISTS).add(date);
                } else if (prType == PrType.MOBILENOTINBOOK) {
                    prlists.get(MOBILENOTINBOOKINCOMELISTS).add(date);
                } else if (prType == PrType.FIXINBOOK) {
                    prlists.get(FIXPHONEINBOOKINCOMEPRLISTS).add(date);
                } else if (prType == PrType.FIXNOTINBOOK) {
                    prlists.get(FIXPHONENOTINBOOKINCOMEPRLISTS).add(date);
                } else {
                    prlists.get(INCOMINGPRLISTS).add(date);
                }
                break;
            case "REJECTED":
                if (prType == PrType.MOBILEINBOOK) {
                    prlists.get(MOBILEINBOOKREJECTEDPRLISTS).add(date);
                } else if (prType == PrType.MOBILENOTINBOOK) {
                    prlists.get(MOBILENOTINBOOKREJECTEDPRLISTS).add(date);
                } else if (prType == PrType.FIXINBOOK) {
                    prlists.get(FIXPHONEINBOOKREJECTEDPRLISTS).add(date);
                } else if (prType == PrType.FIXNOTINBOOK) {
                    prlists.get(FIXPHONENOTINBOOKREJECTEDPRLISTS).add(date);
                } else {
                    prlists.get(REJECTEDPRLISTS).add(date);
                }
                break;
            case "MISSED":
                if (prType == PrType.MOBILEINBOOK) {
                    prlists.get(MOBILEINBOOKMISSEDPRLISTS).add(date);
                } else if (prType == PrType.MOBILENOTINBOOK) {
                    prlists.get(MOBILENOTINBOOKMISSEDPRLISTS).add(date);
                } else if (prType == PrType.FIXINBOOK) {
                    prlists.get(FIXPHONEINBOOKMISSEDPRLISTS).add(date);
                } else if (prType == PrType.FIXNOTINBOOK) {
                    prlists.get(FIXPHONENOTINBOOKMISSEDPRLISTS).add(date);
                } else {
                    prlists.get(MISSEDPRLISTS).add(date);
                }
                break;
            case "BLOCKED":
                if (prType == PrType.MOBILEINBOOK) {
                    prlists.get(MOBILEINBOOKBLOCKEDPRLISTS).add(date);
                } else if (prType == PrType.MOBILENOTINBOOK) {
                    prlists.get(MOBILENOTINBOOKBLOCKEDPRLISTS).add(date);
                } else if (prType == PrType.FIXINBOOK) {
                    prlists.get(FIXPHONEINBOOKBLOCKEDPRLISTS).add(date);
                } else if (prType == PrType.FIXNOTINBOOK) {
                    prlists.get(FIXPHONENOTINBOOKBLOCKEDPRLISTS).add(date);
                } else {
                    prlists.get(BLOCKEDPRLISTS).add(date);
                }
                break;
            default: // UNKNOWN
                if (prType == PrType.MOBILEINBOOK) {
                    prlists.get(MOBILEINBOOKUNKNOWNPRLISTS).add(date);
                } else if (prType == PrType.MOBILENOTINBOOK) {
                    prlists.get(MOBILENOTINBOOKUNKNOWNPRLISTS).add(date);
                } else if (prType == PrType.FIXINBOOK) {
                    prlists.get(FIXPHONEINBOOKUNKNOWNPRLISTS).add(date);
                } else if (prType == PrType.FIXNOTINBOOK) {
                    prlists.get(FIXPHONENOTINBOOKUNKNOWNPRLISTS).add(date);
                } else {
                    prlists.get(UNKNOWNPRLISTS).add(date);
                }
                break;
        }
    }

    private enum PrType {
        MOBILEINBOOK,
        MOBILENOTINBOOK,
        FIXINBOOK,
        FIXNOTINBOOK,
        DEFAULT
    }

    private static Map<String, Set<String>> initPRlists() {
        HashMap<String, Set<String>> result = new HashMap<>();
        result.put(ALLPRLISTS, new HashSet<>());
        result.put(MOBILEPRLISTS, new HashSet<>());
        result.put(MOBILEINBOOKOUTGOPRLISTS, new HashSet<>());
        result.put(MOBILEINBOOKINCOMELISTS, new HashSet<>());
        result.put(MOBILEINBOOKREJECTEDPRLISTS, new HashSet<>());
        result.put(MOBILEINBOOKMISSEDPRLISTS, new HashSet<>());
        result.put(MOBILEINBOOKBLOCKEDPRLISTS, new HashSet<>());
        result.put(MOBILEINBOOKUNKNOWNPRLISTS, new HashSet<>());
        result.put(MOBILENOTINBOOKOUTGOPRLISTS, new HashSet<>());
        result.put(MOBILENOTINBOOKINCOMELISTS, new HashSet<>());
        result.put(MOBILENOTINBOOKREJECTEDPRLISTS, new HashSet<>());
        result.put(MOBILENOTINBOOKMISSEDPRLISTS, new HashSet<>());
        result.put(MOBILENOTINBOOKBLOCKEDPRLISTS, new HashSet<>());
        result.put(MOBILENOTINBOOKUNKNOWNPRLISTS, new HashSet<>());
        result.put(FIXPHONEPRLISTS, new HashSet<>());
        result.put(FIXPHONEINBOOKOUTGOPRLISTS, new HashSet<>());
        result.put(FIXPHONEINBOOKINCOMEPRLISTS, new HashSet<>());
        result.put(FIXPHONEINBOOKREJECTEDPRLISTS, new HashSet<>());
        result.put(FIXPHONEINBOOKMISSEDPRLISTS, new HashSet<>());
        result.put(FIXPHONEINBOOKBLOCKEDPRLISTS, new HashSet<>());
        result.put(FIXPHONEINBOOKUNKNOWNPRLISTS, new HashSet<>());
        result.put(FIXPHONENOTINBOOKOUTGOPRLISTS, new HashSet<>());
        result.put(FIXPHONENOTINBOOKINCOMEPRLISTS, new HashSet<>());
        result.put(FIXPHONENOTINBOOKREJECTEDPRLISTS, new HashSet<>());
        result.put(FIXPHONENOTINBOOKMISSEDPRLISTS, new HashSet<>());
        result.put(FIXPHONENOTINBOOKBLOCKEDPRLISTS, new HashSet<>());
        result.put(FIXPHONENOTINBOOKUNKNOWNPRLISTS, new HashSet<>());
        result.put(PHONEBOOKPRLISTS, new HashSet<>());
        result.put(NOTPHONEBOOKPRLISTS, new HashSet<>());
        result.put(OUTGOINGPRLISTS, new HashSet<>());
        result.put(INCOMINGPRLISTS, new HashSet<>());
        result.put(REJECTEDPRLISTS, new HashSet<>());
        result.put(MISSEDPRLISTS, new HashSet<>());
        result.put(BLOCKEDPRLISTS, new HashSet<>());
        result.put(UNKNOWNPRLISTS, new HashSet<>());

        return result;
    }

    private static JSONObject initPropJson(String propFile) throws IOException {
        InputStream resourceAsStream =
                PhoneRecordLabelAddBakKfkSrc.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        JSONObject res = new JSONObject();
        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
        while (keysIterator.hasNext()) {
            res.put((String) keysIterator.next(), -1);
        }
        return res;
    }
}
